package com.sdleyou.treepro.common.utils;

import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.Map;
import java.util.concurrent.TimeUnit;

@Service
public class RedisDelayedQueue {

    /**
     * 任务回调监听
     *
     * @param <T>
     */
    public abstract static class TaskEventListener<T> {
        /**
         * 执行方法
         *
         * @param t
         */
        public abstract void invoke(T t);
    }

    @Autowired
    private RedissonClient redissonClient;

    /**
     * 添加队列
     *
     * @param t        DTO传输类
     * @param delay    时间数量
     * @param timeUnit 时间单位
     * @param <T>      泛型
     */
    public <T> void addQueue(T t, long delay, TimeUnit timeUnit, TaskEventListener taskEventListener) {
        String queueName = t.getClass().getName();
        RBlockingQueue<T> blockingFairQueue = redissonClient.getBlockingQueue(queueName);
        RDelayedQueue<T> delayedQueue = redissonClient.getDelayedQueue(blockingFairQueue);
        delayedQueue.offer(t, delay, timeUnit);
        delayedQueue.destroy();
        getQueue(queueName, taskEventListener);
    }

    /**
     * 获取队列
     *
     * @param className         className
     * @param taskEventListener 任务回调监听
     * @param <T>               泛型
     * @return
     */
    public synchronized <T> void getQueue(String className, TaskEventListener taskEventListener) {
        RBlockingQueue<T> blockingFairQueue = redissonClient.getBlockingQueue(className);
        //由于此线程需要常驻，可以新建线程，不用交给线程池管理
        Thread repeatThread = null;
        Map<Thread, StackTraceElement[]> threads = Thread.getAllStackTraces();
        for (Thread thread : threads.keySet()) {
            if (thread.getName().equals(className)) {
                repeatThread = thread;
            }
        }
        if (repeatThread == null) {
            Thread thread = new Thread(() -> {
                while (true) {
                    try {
                        T t = blockingFairQueue.take();
                        taskEventListener.invoke(t);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
            thread.setName(className);
            thread.start();
        }
    }
}